package com.mall.order.mq.delay.consumer;

import com.mall.order.dal.entitys.Order;
import com.mall.order.dal.entitys.OrderItem;
import com.mall.order.dal.entitys.Stock;
import com.mall.order.dal.persistence.OrderItemMapper;
import com.mall.order.dal.persistence.OrderMapper;
import com.mall.order.dal.persistence.StockMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import tk.mybatis.mapper.entity.Example;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;

@Slf4j
@Component
public class DelayCancelConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    OrderMapper orderMapper;

    @Autowired
    OrderItemMapper orderItemMapper;

    @Autowired
    StockMapper stockMapper;


    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer35");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        try {
            consumer.subscribe("delay_order_cancel", "*");

            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    // 1. 获取消息
                    MessageExt message = msgs.get(0);
                    //2. 获取orderId
                    try {
                        byte[] body = message.getBody();
                        String orderIdStr = new String(body, 0, body.length, "utf-8");
                        long orderId = Long.parseLong(orderIdStr);
                        log.info("DelayCancelConsumer，获取orderId：" + orderId);
                        //3. 访问数据库，检查订单的支付状态:
                        Example orderExample = new Example(Order.class);
                        orderExample.createCriteria().andEqualTo("orderId", orderId);
                        List<Order> orders = orderMapper.selectByExample(orderExample);
                        Order order = orders.get(0);
                        //   a. 如果订单已支付或者订单已取消。什么都不做
                        if (order.getStatus() == 1 || order.getStatus() == 7) {
                            log.info("DelayCancelConsumer，订单已支付！");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        //   b. 如果发现订单处于一个初始化状态，意味着订单没有被支付，执行取消订单的逻辑
                        if (order.getStatus() == 0) {
                            //      1) 修改订单状态为取消状态
                            order.setStatus(7);
                            order.setUpdateTime(new Date());
                            int affectedRowsOrder = orderMapper.updateByPrimaryKey(order);
                            log.info("DelayCancelConsumer，订单已经被取消，affectedRows:" + affectedRowsOrder);
                            //      2) 扫描订单中的每一个订单商品条目，根据订单中商品条目下单的数量，+stock_count -stock_count
                            // tb_order_item更新释放库存
                            Example orderItemExample = new Example(OrderItem.class);
                            orderItemExample.createCriteria().andEqualTo("orderId", orderId);
                            List<OrderItem> orderItems = orderItemMapper.selectByExample(orderItemExample);
                            for (OrderItem orderItem : orderItems) {
                                orderItem.setStatus(2);
                                int affectedRowsOrderItem = orderItemMapper.updateByPrimaryKey(orderItem);
                                log.info("DelayCancelConsumer，更新释放item库存：" + orderItem + ",affectedRows:" + affectedRowsOrderItem);
                            }

                            // tb_stock恢复锁定的库存
                            for (OrderItem orderItem : orderItems) {
                                Long itemId = orderItem.getItemId();
                                Integer num = orderItem.getNum();
                                // 根据itemId查到锁定库存和实际库存
                                Example stockExample = new Example(Stock.class);
                                stockExample.createCriteria().andEqualTo("itemId", itemId);
                                List<Stock> stocks = stockMapper.selectByExample(stockExample);
                                Stock stock = stocks.get(0);
                                log.info("DelayCancelConsumer，恢复锁定的库存前stock：" + stock);
                                if (stock == null) {
                                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                }
                                stock.setLockCount(stock.getLockCount() - num);
                                stock.setStockCount(stock.getStockCount() + num);
                                int updateByPrimaryKeyAffectedRows = stockMapper.updateStockAndLockCount(stock);
                                log.info("DelayCancelConsumer，恢复锁定的库存后stock：" + stock + ",updateByPrimaryKeyAffectedRows:" + updateByPrimaryKeyAffectedRows);
                            }
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }

                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        log.error("DelayCancelConsumer.init，", e.getMessage());
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 启动消费者
            consumer.start();
            log.info("DefaultMQPushConsumer,消费者启动");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


}
